use ntex::web::{self, HttpResponse};
use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use sse::Broadcaster;
use std::sync::Arc;
use futures::StreamExt;

mod sse;

#[derive(Serialize, Deserialize)]
pub struct ChatRequest {
    pub id: String,
    pub message: String,
}

#[derive(Serialize, Deserialize)]
pub struct ChatResponse {
    pub message: String,
}

async fn init_chat(
    session_id: web::types::Path<String>,
    data: web::types::State<AppState>,
) -> HttpResponse {

    let session_id_inner = session_id.into_inner();
    println!("init: session_id: {}", &session_id_inner);

    let rx = data.broadcaster.lock().new_client();

    HttpResponse::Ok()
        .header("content-type", "text/event-stream")
        .no_chunking()
        .streaming(rx)
}

const MAX_SIZE: usize = 262_144; // 256k

async fn send_message(
    data: web::types::State<AppState>,
    //req: web::types::Json<ChatRequest>,
    mut payload: web::types::Payload
) -> Result<HttpResponse, ntex::web::Error> {
    let mut body = ntex::util::BytesMut::new();
    while let Some(chunk) = payload.next().await {
        let chunk = chunk?;
        // limit max size of in-memory payload
        if (body.len() + chunk.len()) > MAX_SIZE {
            return Err(ntex::web::error::ErrorBadRequest("overflow").into());
        }
        body.extend_from_slice(&chunk);
    }

    // body is loaded, now we can deserialize serde-json
    // let obj = serde_json::from_slice::<ChatRequest>(&body).unwrap();
    // let session_id = &obj.id;
    // let message = &obj.message;
    //println!("session_id: {}, message: {}", session_id, message);
    println!("{}", String::from_utf8_lossy(&body));
    for response in vec!["aaaaa", "bbbbb"] {
        data.broadcaster.lock().send(&response);
    }

    Ok(HttpResponse::Ok().json(&ChatResponse {
        message: "msg sent".to_string(),
    }))
}

struct AppState {
    broadcaster: Arc<Mutex<Broadcaster>>,
}

#[ntex::main]
async fn main() -> std::io::Result<()> {
    env_logger::init_from_env(env_logger::Env::default().default_filter_or("info"));


    let broadcaster = Broadcaster::create();

    println!("Starting server at http://127.0.0.1:8080");

    web::HttpServer::new(move || {
        web::App::new()
            .wrap(web::middleware::Logger::default())
            .wrap(ntex_cors::Cors::default())
            .state(AppState {
                broadcaster: broadcaster.clone(),
            })
            //.service(web::resource("/").to(index))
            .route("/api/init/{session_id}", web::get().to(init_chat))
            .route("/api/chat", web::post().to(send_message))
    })
    .bind(("0.0.0.0", 8080))?
    .run()
    .await
}
